package qy.qyalgotrader.wind;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.NumberFormat;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;

import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.math3.stat.StatUtils;
import org.apache.commons.math3.stat.descriptive.moment.Mean;
import org.apache.commons.math3.stat.descriptive.summary.Sum;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.math.DoubleMath;
import com.mongodb.client.AggregateIterable;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.Accumulators;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Indexes;
import com.mongodb.client.model.Sorts;

import it.unimi.dsi.fastutil.ints.IntArrayList;

import static com.mongodb.client.model.Filters.*;

import qy.jalgotrade.bar.Bar;
import qy.jalgotrade.bar.Bar.Frequency;
import qy.jalgotrade.bar.BasicBar;
import qy.jalgotrade.utils.CommonUtils;
import qy.jalgotrade.Initializer;
import qy.qyalgotrader.utils.CommonDataUtils;
import qy.qyalgotrader.utils.Constants;
import qy.qyalgotrader.utils.Helper;
import qy.qyalgotrader.utils.Services;
import qy.qyalgotrader.utils.apiservice.wind.Wapi;
import qy.qyalgotrader.utils.apiservice.wind.WapiPrx;
import qy.qyalgotrader.utils.apiservice.wind.WindData;
import tech.tablesaw.api.ColumnType;
import tech.tablesaw.api.DateTimeColumn;
import tech.tablesaw.api.DoubleColumn;
import tech.tablesaw.api.Row;
import tech.tablesaw.api.Table;
import tech.tablesaw.columns.Column;
import tech.tablesaw.io.csv.CsvReadOptions;
//import tech.tablesaw.io.csv.CsvWriteOptions;
import tech.tablesaw.selection.Selection;

/**
 * @author qy
 *
 */
public final class WindDataUtils {

	private static final Logger logger = LoggerFactory.getLogger(WindDataUtils.class);

	/**
	 * ${...}/my_quant/wind_hisdata
	 */
	public static final Path WIND_HIS_DATA_BASE_DIR_DEFAULT = Paths
	        .get(Initializer.APP_DEPLOY_CONFIG.getSection("my_quants").getString("windHisDataBaseDir"));

	/**
	 * ${...}/my_quant/wind_hisdata/tick
	 */
	public static final Path WIND_TICK_CSV_BASE_DEFAULT = WIND_HIS_DATA_BASE_DIR_DEFAULT.resolve("tick");

	/**
	 * qyalgotrader MongoDB name:
	 */
	private static final String __DB_NAME = "qyalgodb";

	private static final MongoClient __mongoClient = Services.getInstance().getMongoClient();

	/**
	 * waiting time (sec.) calling w.start():
	 */
	public static int WIND_CONNECT_WAIT_TIME = 5;

	/**
	 * Wapi:
	 */
	private static WapiPrx __wapi;

	static {

		try {
			__wapi = WapiPrx.checkedCast(Services.getInstance().createIceWapiProxy("Wapi"));
		} catch (Exception e) {
			logger.error("Exception occured initializing Wapi: {}", ExceptionUtils.getStackTrace(e));
		}
	}

	/**
	 * retry couut: 3
	 * @throws Exception
	 */
	public static void ensureWindConnected() throws Exception {

		ensureWindConnected(3);
	}

	/**
	 * 
	 * @param retry retry couut: 3 (default)
	 * @throws Exception
	 */
	public static void ensureWindConnected(int retry) throws Exception {

		assert retry > 0;
		WindData wd = null;
		while (retry-- > 0) {
			if (__wapi.isconnected()) {
				return;
			} else {
				wd = __wapi.start(OptionalLong.of(WIND_CONNECT_WAIT_TIME * 1000));
				if (wd.errorCode == 0) {
					return;
				}
			}
		}
		if (wd.errorCode != 0) {
			throw new Exception(String.format("Error starting WindPy (w.start()): %s", wd));
		}
	}

	/**
	 * daysOffset >= 0; 获取 startDay 起始的前 daysOffset 个交易日:
	 * 
	 * @param startDay startDay
	 * @param daysOffset daysOffset
	 * @return
	 */
	public static LocalDate tradingDaysOffset(LocalDate startDay, int daysOffset) {

		return getTradingDays(startDay, daysOffset, true).get(0);
	}

	/**
	 * <p>
	 * 以 startDay 为起点, 获取前 days (>= 0, 若 days = 0, 则返回 当前交易日) 个交易日 (若 inclZeroTd 为 False (默认), 则返回不含 起始交易日,
	 * 即 offset 为 0 的交易日):
	 * </p>
	 * <p>
	 * 注意: 与 qyalgotrader.ts.barfeed.get_trading_days() 不同: 当 startDay 为非交易日时, days 为 0 时返回最近的一个交易日,
	 * days 为 1 时返回最近交易日的上个交易日, 以此类推.
	 * </p>
	 * 
	 * @param startDay startDay
	 * @param days days
	 * @return
	 */
	public static List<LocalDate> getTradingDays(LocalDate startDay, int days) {

		return getTradingDays(startDay, days, false);
	}

	/**
	 * <p>
	 * 以 startDay 为起点, 获取前 days (>= 0, 若 days = 0, 则返回 当前交易日) 个交易日 (若 inclZeroTd 为 False, 则返回不含 起始交易日, 即
	 * offset 为 0 的交易日):
	 * </p>
	 * <p>
	 * 注意: 与 qyalgotrader.ts.barfeed.get_trading_days() 不同: 当 startDay 为非交易日时, days 为 0 时返回最近的一个交易日,
	 * days 为 1 时返回最近交易日的上个交易日, 以此类推.
	 * </p>
	 * 
	 * @param startDay startDay
	 * @param days days
	 * @param inclZeroTd inclZeroTd
	 * @return
	 */
	public static List<LocalDate> getTradingDays(LocalDate startDay, int days, boolean inclZeroTd) {

		assert days >= 0;

		List<LocalDate> tds = null;
		String collName = "qset.tdays";
		String dtStr = startDay.format(Constants.DATE_FORMAT_STD);
		/*
		 * select max(date) as date
		 * from "qset.tdays":
		 */
		MongoCollection<Document> coll = __mongoClient.getDatabase(__DB_NAME).getCollection(collName);
		AggregateIterable<Document> cursAgg = coll
		        .aggregate(Arrays.asList(Aggregates.group(null, Accumulators.max("date", "$date"))));
		LocalDate maxDate = LocalDate.parse(cursAgg.first().getString("date"), Constants.DATE_FORMAT_STD);

		if (startDay.compareTo(maxDate) <= 0) {

			tds = new ArrayList<>();
			/*
			 * select *
			 * from "qset.tdays"
			 * where "SHFE" == True,
			 *     "date" <= dtStr
			 * order by "date" DESC
			 */
			FindIterable<Document> curs = coll.find(and(eq("SHFE", true), lte("date", dtStr)))
			        .hint(Indexes.ascending("date")).sort(Sorts.descending("date"));

			for (Document e : curs.limit(days + 1)) {
				tds.add(LocalDate.parse(e.getString("date"), Constants.DATE_FORMAT_STD));
			}

			Collections.reverse(tds);

			if (days == 0) {
				return tds.subList(tds.size() - 1, tds.size());
			} else {
				if (inclZeroTd) {
					return tds;
				} else {
					return tds.subList(0, tds.size() - 1);
				}
			}
		} else {
			try {
				ensureWindConnected();
				Wapi.TdaysoffsetResult ret = __wapi.tdaysoffset(-days, dtStr, Optional.empty());
				if (ret.wd.errorCode != 0) {
					logger.error("Call w.tdaysoffset({}, {}, \"TradingCalendar=SHFE\") with errors: {}", -days, dtStr,
							Helper.toString(ret.wd));
					return Collections.emptyList();
				} else {
					tds = new ArrayList<>();
					WindData wd = __wapi.tdays(ret.returnValue, dtStr, Optional.empty());
					if (wd.errorCode != 0) {
						logger.error("Call w.tdays({}, {}, \"TradingCalendar=SHFE\") with errors: {}", ret.returnValue,
								dtStr, Helper.toString(ret.wd));
					} else {
						for (String e : wd.times) {
							tds.add(LocalDate.parse(e, Constants.DATE_FORMAT_STD));
						}
					}
				}
			} catch (Exception e) {
				logger.error("Exception occured: {}", ExceptionUtils.getStackTrace(e));
				return Collections.emptyList();
			}
		}

		return tds;
	}

	/**
	 * 
	 * @param fromDate fromDate
	 * @param toDate toDate
	 * @return
	 */
	public static List<LocalDate> tradingDaysBy(LocalDate fromDate, LocalDate toDate) {

		assert toDate.compareTo(fromDate) >= 0;

		List<LocalDate> tds = null;
		String collName = "qset.tdays";
		String fromDtStr = fromDate.format(Constants.DATE_FORMAT_STD);
		String toDtStr = toDate.format(Constants.DATE_FORMAT_STD);
		/*
		 * select max(date) as date
		 * from "qset.tdays":
		 */
		MongoCollection<Document> coll = __mongoClient.getDatabase(__DB_NAME).getCollection(collName);
		AggregateIterable<Document> cursAgg = coll
		        .aggregate(Arrays.asList(Aggregates.group(null, Accumulators.max("date", "$date"))));
		LocalDate maxDate = LocalDate.parse(cursAgg.first().getString("date"), Constants.DATE_FORMAT_STD);

		if (toDate.compareTo(maxDate) <= 0) {

			tds = new ArrayList<>();
			/*
			 * select *
			 * from "qset.tdays"
			 * where "SHFE" == True,
			 *     "date" >= toDtStr and "date" <= fromDtStr
			 * order by "date" ASC
			 */
			FindIterable<Document> curs = coll.find(and(eq("SHFE", true), lte("date", toDtStr), gte("date", fromDtStr)))
			        .hint(Indexes.ascending("date")).sort(Sorts.ascending("date"));
			for (Document e : curs) {
				tds.add(LocalDate.parse(e.getString("date"), Constants.DATE_FORMAT_STD));
			}
		} else {
			try {
				ensureWindConnected();
				WindData wd = __wapi.tdays(fromDtStr, toDtStr, Optional.empty());
				if (wd.errorCode != 0) {
					logger.error("Call w.tdays({}, {}, \"TradingCalendar=SHFE\") with errors: {}", fromDtStr, toDtStr,
							Helper.toString(wd));
					return Collections.emptyList();
				} else {
					tds = new ArrayList<>();
					for (String e : wd.times) {
						tds.add(LocalDate.parse(e, Constants.DATE_FORMAT_STD));
					}
				}
			} catch (Exception e) {
				logger.error("Exception occured: {}", ExceptionUtils.getStackTrace(e));
				return Collections.emptyList();
			}
		}

		return tds;
	}

	/**
	 * 读取 wind 历史 Tick csv
	 * 
	 * @param wc wc
	 * @param theDate theDate
	 * @param tickCsvBaseP default: TS_HIS_DATA_BASE_DIR_DEFAULT
	 * @param csvFileNameExtP default: ".csv"
	 * @param csvFileEncP default: "UTF-8"
	 * @return
	 * @throws IOException
	 */
	public static Table readFromTicksCsvWind(String wc, LocalDate theDate, Optional<Path> tickCsvBaseP,
			Optional<String> csvFileNameExtP, Optional<String> csvFileEncP) throws IOException {

		Path tickCsvBase = tickCsvBaseP.orElse(WIND_TICK_CSV_BASE_DEFAULT);
		String csvFileNameExt = csvFileNameExtP.orElse(".csv");
		String csvFileEnc = csvFileEncP.orElse("UTF-8");

		return readFromTicksCsvWind(wc, theDate, tickCsvBase, csvFileNameExt, csvFileEnc);
	}

	/**
	 * 读取 wind 历史 Tick csv
	 * 
	 * @param wc wc
	 * @param theDate theDate
	 * @param tickCsvBase tickCsvBase
	 * @param csvFileNameExt csvFileNameExt
	 * @param csvFileEnc csvFileEnc
	 * @return
	 * @throws IOException
	 */
	public static Table readFromTicksCsvWind(String wc, LocalDate theDate, Path tickCsvBase, String csvFileNameExt,
			String csvFileEnc) throws IOException {

		String dtStr = theDate.format(Constants.DATE_FORMAT_YYYYMMDD);
		String dtYearStr = theDate.format(Constants.DATE_FORMAT_YYYY);
		Path tickCsvP = tickCsvBase.resolve(dtYearStr).resolve(dtStr).resolve(wc + csvFileNameExt);
		Table df = null;
		if (tickCsvP.toFile().exists()) {
			try (InputStreamReader reader = new InputStreamReader(new FileInputStream(tickCsvP.toFile()),
			        Charset.forName(csvFileEnc))) {
				// @formatter:off
				Map<String, ColumnType> colsTypeMap = MapUtils.putAll(new HashMap<>(), new Object[][] {
					{ Constants.DATE_TIME, ColumnType.LOCAL_DATE_TIME },
					{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("pre_close", "resp"), ColumnType.DOUBLE },
					{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("last", "resp"), ColumnType.DOUBLE },
					{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("volume", "resp"), ColumnType.DOUBLE },
					{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("amt", "resp"), ColumnType.DOUBLE },
					{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("oi", "resp"), ColumnType.DOUBLE }
				});
				ColumnType[] types = CommonUtils.genColumnsInfoFor(tickCsvP, colsTypeMap, csvFileEnc);
				// @formatter:on
				CsvReadOptions options = CsvReadOptions.builder(reader).missingValueIndicator("").columnTypes(types)
				        .dateTimeFormat(Constants.DATE_TIME_FORMAT_MS).build();
	
				df = Table.read().usingOptions(options);
			}
			
			if (df.isEmpty()) {
				logger.info("empty wind tick data file: {} .", tickCsvP);
			}
		} else {
			logger.info("wind tick data file: {} does NOT exists !", tickCsvP);
		}
		return df;
	}

	/**
	 * <pre>
	 * 1. 过滤掉 datetime 不在交易时间范围内的 tick;
	 * 2. 过滤重复 Tick (仅盘口有变动无成交的 Tick, 郑商所收盘 Tick, ...).
	 * </pre>
	 * 
	 * @param ticksDf ticksDf
	 * @param wc wc
	 * @return
	 */
	public static Table purifyTicksDataFrame(Table ticksDf, String wc) {

		// 过滤掉 datetime 不在交易时间范围内的 tick:
		IntArrayList indxes = new IntArrayList();
		ticksDf.forEach(row -> {
			if (CommonDataUtils.isIllegalCnFuTickDatetime(wc, row.getDateTime(Constants.DATE_TIME),
					OptionalInt.empty(), Optional.empty(), Optional.empty(), Optional.empty())) {
				indxes.add(row.getRowNumber());
			}
		});
		if (indxes.size() > 0) {
			ticksDf = ticksDf.dropRows(indxes.toIntArray());
		}
		// 过滤重复 Tick (仅盘口有变动无成交的 Tick, 郑商所收盘 Tick, ...):
		double lstVol = 0.0;
		IntArrayList indxes1 = new IntArrayList();
		for (Row row : ticksDf) {
			double v = row.getDouble(
			        (String) Constants.WIND_FIELD_NAME_MAPPING_WST.get("volume", "resp"));
			if (DoubleMath.fuzzyEquals(v, lstVol, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH)) {
				indxes1.add(row.getRowNumber());
			}
			lstVol = v;
		}
		if (indxes1.size() > 0) {
			ticksDf = ticksDf.dropRows(indxes1.toIntArray());
		}
		return ticksDf;
	}

	/**
	 * 读取 wind 历史 minutes csv
	 * 
	 * @param wc wc
	 * @param theDate theDate
	 * @param freq freq
	 * @param minsCsvBaseP default: WIND_HIS_DATA_BASE_DIR_DEFAULT
	 * @param csvFileNameExtP default: ".csv"
	 * @param csvFileEncP default: "UTF-8"
	 * @return
	 * @throws IOException
	 */
	public static Table readFromMinutesCsvWind(String wc, LocalDate theDate, Frequency freq, Optional<Path> minsCsvBaseP,
			Optional<String> csvFileNameExtP, Optional<String> csvFileEncP) throws IOException {

		Path minsCsvBase = minsCsvBaseP.orElse(WIND_HIS_DATA_BASE_DIR_DEFAULT);
		String csvFileNameExt = csvFileNameExtP.orElse(".csv");
		String csvFileEnc = csvFileEncP.orElse("UTF-8");

		return readFromMinutesCsvWind(wc, theDate, freq, minsCsvBase, csvFileNameExt, csvFileEnc);
	}

	/**
	 * 读取 wind 历史 minutes csv
	 * 
	 * @param wc wc
	 * @param theDate theDate
	 * @param freq freq
	 * @param minsCsvBase minsCsvBase
	 * @param csvFileNameExt csvFileNameExt
	 * @param csvFileEnc csvFileEnc
	 * @return
	 * @throws IOException
	 */
	public static Table readFromMinutesCsvWind(String wc, LocalDate theDate, Frequency freq, Path minsCsvBase,
			String csvFileNameExt, String csvFileEnc) throws IOException {

		String dtStr = theDate.format(Constants.DATE_FORMAT_YYYYMMDD);
		String dtYearStr = theDate.format(Constants.DATE_FORMAT_YYYY);
		Path minsCsvP = minsCsvBase.resolve(CommonDataUtils.barFreqToHisDataDir(freq))
				.resolve(dtYearStr).resolve(dtStr).resolve(wc + csvFileNameExt);
		Table df = null;
		if (minsCsvP.toFile().exists()) {
			try (InputStreamReader reader = new InputStreamReader(new FileInputStream(minsCsvP.toFile()),
			        Charset.forName(csvFileEnc))) {
				// @formatter:off
				Map<String, ColumnType> colsTypeMap = MapUtils.putAll(new HashMap<>(), new Object[][] {
					{ Constants.DATE_TIME, ColumnType.LOCAL_DATE_TIME },
					{ Constants.BAR_FIELD_OPEN, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_CLOSE, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_HIGH, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_LOW, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_VOLUME, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_AMOUNT, ColumnType.DOUBLE }
				});
				ColumnType[] types = CommonUtils.genColumnsInfoFor(minsCsvP, colsTypeMap, csvFileEnc);
				// @formatter:on
				CsvReadOptions options = CsvReadOptions.builder(reader).missingValueIndicator("").columnTypes(types)
				        .dateTimeFormat(Constants.DATE_TIME_FORMAT_STD).build();

				df = Table.read().usingOptions(options);
			}
			if (df.isEmpty()) {
				logger.info("empty wind minutes({}) data file: {} .", freq, minsCsvP);
			}
		} else {
			logger.info("wind minutes({}) data file: {} does NOT exists !", freq, minsCsvP);
		}
		return df;
	}

	/**
	 * Ticks 合成 minute(s) Bar:
	 * 
	 * @param codeOrWc codeOrWc
	 * @param td td
	 * @param ticksDf ticksDf
	 * @param startDm startDm
	 * @param endDm endDm
	 * @param cnFuTdInfos cnFuTdInfos
	 * @param lstTkDelay lstTkDelay
	 * @return
	 */
	public static Table sampleDataFrameTicksToMinuteWind(String codeOrWc, LocalDate td, Table ticksDf,
	        LocalDateTime startDm, LocalDateTime endDm, Map<String, Object> cnFuTdInfos, int lstTkDelay) {

		String exchId = CommonDataUtils.getExchangeIdByCnFuturesCode(codeOrWc);
		// minute 级别周期 df:
		List<Column<?>> smplCols = new ArrayList<>();
		DateTimeColumn dmCol = DateTimeColumn.create(Constants.DATE_TIME);
		dmCol.setPrintFormatter(Constants.DATE_TIME_FORMAT_STD);
		smplCols.add(dmCol);
		for (String e : Constants.DEFAULT_FIELDS_MIN_1) {
			DoubleColumn doubleCol = DoubleColumn
			        .create((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get(e, "resp"));
			if (e.equals(Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("volume", "resp"))
			        || e.equals(Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("amt", "resp"))
			        || e.equals(Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("oi", "amt"))) {
				// 避免使用科学计数格式:
				NumberFormat fmt = NumberFormat.getNumberInstance();
				fmt.setGroupingUsed(false);
				doubleCol.setPrintFormatter(fmt, "");
			}
			smplCols.add(doubleCol);
		}
		Table smplDf = Table.create(smplCols);

		// 用来计算 bar 的 涨跌 及 涨跌幅, 每新产生一个 bar 后更新, 初始值为前收盘价 (从 tickDf 中获取):
		double priceRefVal = ticksDf.row(0)
		        .getDouble((String) Constants.WIND_FIELD_NAME_MAPPING_WST.get("pre_close", "resp"));
		String lastPricsField = (String) Constants.WIND_FIELD_NAME_MAPPING_WST.get("last", "resp");
		String volField = (String) Constants.WIND_FIELD_NAME_MAPPING_WST.get("volume", "resp");
		String amtField = (String) Constants.WIND_FIELD_NAME_MAPPING_WST.get("amt", "resp");
		String posField = (String) Constants.WIND_FIELD_NAME_MAPPING_WST.get("oi", "resp");
		// 上一个 Bar 当日累计 Vol, Amount, Position (即 wst() 返回的上个 Slice 的 vol, amt 和 oi 值):
		double lastSliceAccuVol = Double.NaN;
		double lastSliceAccuAmt = Double.NaN;

		LocalDateTime sliceStartDm = startDm;
		while (sliceStartDm.compareTo(endDm) < 0) {
			LocalDateTime sliceEndDm = sliceStartDm.plusSeconds(Frequency.MINUTE.getValue());
			// 跳过非交易时间:
			if (CommonDataUtils.isIllegalCnFuTickDatetime(codeOrWc, sliceStartDm, OptionalInt.of(0), Optional.empty(),
					Optional.empty(), Optional.empty())) {
				sliceStartDm = sliceEndDm;
				continue;
			}

			// 连续竞价时段首个 Bar:
			Pair<Boolean, LocalDateTime> firstBarTup = CommonDataUtils.isCnFuContia1stTimeRange(codeOrWc, sliceStartDm,
			        sliceEndDm, Optional.of(td), Optional.of(cnFuTdInfos), Optional.empty());
			// 连续竞价收盘 (交易暂停) 最后一个 Bar (若 sliceEndDm > endDm, 则亦不会丢失最后一个 Tick):
			Pair<Boolean, LocalDateTime> lastBarTup = CommonDataUtils.isCnFuContiaLastTimeRange(codeOrWc, sliceStartDm,
			        sliceEndDm, OptionalInt.of(lstTkDelay), Optional.of(td), Optional.of(cnFuTdInfos), Optional.empty());

			Selection filter = null;
			if (firstBarTup.getLeft()) {

				filter = ticksDf.dateTimeColumn(Constants.DATE_TIME).isOnOrAfter(firstBarTup.getRight())
				        .and(ticksDf.dateTimeColumn(Constants.DATE_TIME).isBefore(sliceEndDm));

			} else if (lastBarTup.getLeft()) {

				filter = ticksDf.dateTimeColumn(Constants.DATE_TIME).isOnOrAfter(sliceStartDm)
				        .and(ticksDf.dateTimeColumn(Constants.DATE_TIME).isBefore(lastBarTup.getRight()));

			} else {

				filter = ticksDf.dateTimeColumn(Constants.DATE_TIME).isOnOrAfter(sliceStartDm)
				        .and(ticksDf.dateTimeColumn(Constants.DATE_TIME).isBefore(sliceEndDm));
			}
			Table ticksSlice = ticksDf.where(filter);

			if (!ticksSlice.isEmpty()) {
				double open = ticksSlice.doubleColumn(lastPricsField).getDouble(0);
				double high = ticksSlice.doubleColumn(lastPricsField).max();
				double low = ticksSlice.doubleColumn(lastPricsField).min();
				double close = ticksSlice.doubleColumn(lastPricsField).getDouble(ticksSlice.rowCount() - 1);
				// 涨跌
				double priceChg = close - priceRefVal;
				// 涨跌幅
				double priceChgPct = !DoubleMath.fuzzyEquals(priceRefVal, 0.0, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH)
				        ? priceChg / priceRefVal
				        : Double.NaN;
				// 持仓量
				double pos = ticksSlice.doubleColumn(posField).getDouble(ticksSlice.rowCount() - 1);

				double ticksSliceLastVol = ticksSlice.doubleColumn(volField).getDouble(ticksSlice.rowCount() - 1);
				double volume = !Double.isNaN(lastSliceAccuVol) ? ticksSliceLastVol - lastSliceAccuVol
				        : ticksSliceLastVol;
				// 补 amount 时使用 (郑商所品种)
				double lastSliceAccuVolBkp = lastSliceAccuVol;
				lastSliceAccuVol = ticksSliceLastVol;

				double tickSliceLastAmt = ticksSlice.doubleColumn(amtField).getDouble(ticksSlice.rowCount() - 1);
				double amount = !Double.isNaN(lastSliceAccuAmt) ? tickSliceLastAmt - lastSliceAccuAmt
				        : tickSliceLastAmt;
				lastSliceAccuAmt = tickSliceLastAmt;
				// 补 amount 时使用 (郑商所品种)
				double subsVol = !Double.isNaN(lastSliceAccuVolBkp) ? lastSliceAccuVolBkp : 0.0;
				double[] tickVolArray = new double[ticksSlice.rowCount()];
				for (int i = 0; i < ticksSlice.rowCount(); i++) {
					tickVolArray[i] = ticksSlice.doubleColumn(volField).getDouble(i) - subsVol;
					subsVol = ticksSlice.doubleColumn(volField).getDouble(i);
				}
				if (exchId.equals("CZCE")
				        && DoubleMath.fuzzyCompare(volume, 0, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH) > 0) {
					// 郑商所品种: tickVolSeri 为每个 Tick 成交量 Series, 成交价格 Series 即 ticksSlice["last"]:
					double[] tickPriceArray = ticksSlice.doubleColumn(lastPricsField).asDoubleArray();
					amount = new Sum().evaluate(tickPriceArray, tickVolArray) * (double) cnFuTdInfos.get("tradeUnit");
				}
				// 更新最近收盘价:
				priceRefVal = close;
				// 当成交量为 0 时 (仅盘口发生变化, 有 Tick), 不生成 Bar:
				if (DoubleMath.fuzzyCompare(volume, 0.0, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH) > 0) {
					Row row = smplDf.appendRow();
					row.setDateTime(Constants.DATE_TIME, sliceStartDm);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("open", "resp"), open);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("close", "resp"), close);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("high", "resp"), high);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("low", "resp"), low);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("volume", "resp"), volume);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("amt", "resp"), amount);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("chg", "resp"), priceChg);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("pct_chg", "resp"), priceChgPct);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("oi", "resp"), pos);
				}
			}
			sliceStartDm = sliceEndDm;
		}

		return smplDf;
	}

	/**
	 * minute 级别 变频 为 resFreq 级别:
	 * 
	 * @param min1Df min1Df
	 * @param startDm startDm
	 * @param endDm endDm
	 * @param resFreq resFreq
	 * @param priceRefVal priceRefVal
	 * @return
	 */
	public static Table resampleDataFrameMinuteToWind(Table min1Df, LocalDateTime startDm, LocalDateTime endDm,
	        Frequency resFreq, double priceRefVal) {

		return resampleDataFrameMinuteToWind(min1Df, startDm, endDm, resFreq, priceRefVal, true);
	}

	/**
	 * minute 级别 变频 为 resFreq 级别:
	 * 
	 * @param min1Df min1Df
	 * @param startDm startDm
	 * @param endDm endDm
	 * @param resFreq resFreq
	 * @param priceRefVal priceRefVal
	 * @param dropEmptyBars dropEmptyBars
	 * @return
	 */
	public static Table resampleDataFrameMinuteToWind(Table min1Df, LocalDateTime startDm, LocalDateTime endDm,
	        Frequency resFreq, double priceRefVal, boolean dropEmptyBars) {

		List<Column<?>> rsmpCols = new ArrayList<>();
		DateTimeColumn dmCol = DateTimeColumn.create(Constants.DATE_TIME);
		dmCol.setPrintFormatter(Constants.DATE_TIME_FORMAT_STD);
		rsmpCols.add(dmCol);
		for (String e : Constants.DEFAULT_FIELDS_MIN_1) {
			DoubleColumn doubleCol = DoubleColumn
			        .create((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get(e, "resp"));
			if (e.equals(Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("volume", "resp"))
			        || e.equals(Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("amt", "resp"))
			        || e.equals(Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("oi", "amt"))) {
				// 避免使用科学计数格式:
				NumberFormat fmt = NumberFormat.getNumberInstance();
				fmt.setGroupingUsed(false);
				doubleCol.setPrintFormatter(fmt, "");
			}
			rsmpCols.add(doubleCol);
		}
		Table rsmpDf = Table.create(rsmpCols);

		LocalDateTime sliceStartDm = startDm;
		while (sliceStartDm.compareTo(endDm) < 0) {
			LocalDateTime sliceEndDm = sliceStartDm.plusSeconds(resFreq.getValue());
			// TimeRange 范围内所有 Bars:
			Selection filter = min1Df.dateTimeColumn(Constants.DATE_TIME).isOnOrAfter(sliceStartDm)
			        .and(min1Df.dateTimeColumn(Constants.DATE_TIME).isBefore(sliceEndDm));
			Table barsSlice = min1Df.where(filter);

			if (!barsSlice.isEmpty()) {
				double open = barsSlice
				        .doubleColumn((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("open", "resp"))
				        .getDouble(0);
				double high = barsSlice
				        .doubleColumn((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("high", "resp")).max();
				double low = barsSlice.doubleColumn((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("low", "resp"))
				        .min();
				double close = barsSlice
				        .doubleColumn((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("close", "resp"))
				        .getDouble(barsSlice.rowCount() - 1);
				double volume = barsSlice
				        .doubleColumn((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("volume", "resp")).sum();
				double amount = barsSlice
				        .doubleColumn((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("amt", "resp")).sum();
				// 涨跌
				double priceChg = close - priceRefVal;
				// 涨跌幅
				double priceChgPct = !DoubleMath.fuzzyEquals(priceRefVal, 0.0, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH)
				        ? priceChg / priceRefVal
				        : Double.NaN;
				double pos = barsSlice.doubleColumn((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("oi", "resp"))
				        .getDouble(barsSlice.rowCount() - 1);
				// 更新最近收盘价:
				priceRefVal = close;
				// 当成交量为 0 时 (仅盘口发生变化, 有 Tick), 不生成 Bar:
				if (volume > 0 || !dropEmptyBars) {
					Row row = rsmpDf.appendRow();
					row.setDateTime(Constants.DATE_TIME, sliceStartDm);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("open", "resp"), open);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("close", "resp"), close);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("high", "resp"), high);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("low", "resp"), low);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("volume", "resp"), volume);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("amt", "resp"), amount);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("chg", "resp"), priceChg);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("pct_chg", "resp"), priceChgPct);
					row.setDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("oi", "resp"), pos);
				}
			}
			sliceStartDm = sliceEndDm;
		}
		return rsmpDf;
	}

	/**
	 * 
	 * @param minsDf minsDf
	 * @param freq freq
	 * @param wc wc
	 * @param td td
	 * @return
	 * @throws Exception
	 */
	public static List<Bar> minutesDataFrameToBarListWind(Table minsDf, Frequency freq,
			String wc, LocalDate td) throws Exception {

		List<Bar> barList = new ArrayList<>();

		Map<String, Object> tdInfos = CommonDataUtils.getTradeInfos(wc, td);
		String exchId = Constants.exchSffxToName(CommonDataUtils.windCodeTuple(wc).getRight());

		Mean meanUtils = new Mean();
		double[] closes = minsDf.doubleColumn(Constants.BAR_FIELD_CLOSE).asDoubleArray();
		double[] vols = minsDf.doubleColumn(Constants.BAR_FIELD_VOLUME).asDoubleArray();
		double[] amts = minsDf.doubleColumn(Constants.BAR_FIELD_AMOUNT).asDoubleArray();
		for (int i = 0; i < minsDf.rowCount(); i++) {
			/*
			 * Bar 均价 (vwap) = amount / volume / tradeUnit:
			 *     1. 沪深交易所指数: 不可用;
			 *     2. 郑商所 wsi min_1 数据: 不可用 (0.0, 由于 amount 为 0).
			 * 日内分时均价 (vwap_intraday):
			 *     1. 沪深交易所指数: close 加权平均 (按 amount 加权);
			 *     2. 郑商所 wsi min_1 数据: close 加权平均 (按 volume 加权).
			 */
			Row aRow = minsDf.row(i);
			ZonedDateTime dm = CommonDataUtils.toMarketDatetime(aRow.getDateTime(Constants.DATE_TIME));
			double open = aRow.getDouble(Constants.BAR_FIELD_OPEN);
			double high = aRow.getDouble(Constants.BAR_FIELD_HIGH);
			double low = aRow.getDouble(Constants.BAR_FIELD_LOW);
			double close = aRow.getDouble(Constants.BAR_FIELD_CLOSE);
			double vol = aRow.getDouble(Constants.BAR_FIELD_VOLUME);
			double amt = aRow.getDouble(Constants.BAR_FIELD_AMOUNT);
			// Bar VWAP = amount / volume / tradeUnit (郑商所计算值为 0.0):
			double vwap = Double.NaN;
			double vwapIntraday = Double.NaN;
			if (DoubleMath.fuzzyEquals(vol, 0.0, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH)) {
				vwap = amt / vol / (double) tdInfos.get("tradeUnit");
			}
			if (Arrays.asList("SSE", "SZSE").contains(exchId) && CommonDataUtils.isSseSzseIndex(wc)) {
				// 沪深交易所指数 Bar vwap 和 vwap_intraday:
				// vwap: NaN
				// vwap_intraday: close 加权平均 (按 amount 加权):
				vwapIntraday = meanUtils.evaluate(closes, amts, 0, i + 1);
			} else if (exchId.equals("CZCE")) {
				// 郑商所 Bar vwap 和 vwap_intraday:
				// vwap: 0.0
				// vwap_intraday: close 加权平均 (按 volume 加权):
				vwapIntraday = meanUtils.evaluate(closes, vols, 0, i + 1);
			} else {
				double amtAccu = StatUtils.sum(amts, 0, i + 1);
				double volAccu = StatUtils.sum(vols, 0, i + 1);
				if (!DoubleMath.fuzzyEquals(volAccu, 0.0, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH)) {
					vwapIntraday = amtAccu / volAccu / (double) tdInfos.get("tradeUnit");
				}
			}
			// 仅 min_1 频率 Bar 计算 vwap_intraday:
			if (!(freq == Frequency.MINUTE || freq == Frequency.MINUTE_1)) {
				vwapIntraday = Double.NaN;
			}

			// @formatter:off
			Bar aBar = new BasicBar(dm, open, high, low, close, vol, close, freq,
			        MapUtils.putAll(new HashMap<>(), new Object[][] {
			        	{ Constants.BAR_FIELD_AMOUNT, amt },
			            { Constants.BAR_FIELD_VWAP, vwap },
			            { Constants.BAR_FIELD_VWAP_INTRADAY, vwapIntraday },
			        }));
			// @formatter:on
			barList.add(aBar);
		}

		if (barList.size() == 0) {
			logger.warn("EMPTY minutes DataFrame ({}) for {} !", freq, wc);
		}

		return barList;
	}

	/**
	 * 仅支持 Tick 级别 resample 到日内级别: MIN_1, MIN_3, MIN_5, MIN_15, MIN_30, MIN_60:
	 * 
	 * @param windCodes windCodes
	 * @param td td
	 * @param resFreq resFreq
	 * @param cnFuNamesP default: Collections.emptyList()
	 * @param baseDirP default: default: WIND_HIS_DATA_BASE_DIR_DEFAULT
	 * @param rewriteP default: false
	 * @param breakOnExceptP default: false
	 * @param genEmptyFileP default: false
	 * @param encP default: "UTF-8"
	 * @throws Exception
	 */
	public static void resampleHisTickDataFileWind(List<String> windCodes, LocalDate td, Frequency resFreq,
	        Optional<List<String>> cnFuNamesP, Optional<Path> baseDirP, Optional<Boolean> rewriteP,
	        Optional<Boolean> breakOnExceptP, Optional<Boolean> genEmptyFileP, Optional<String> encP) throws Exception {

		List<String> cnFuNames = cnFuNamesP.orElse(Collections.emptyList());
		Path baseDir = baseDirP.orElse(WIND_HIS_DATA_BASE_DIR_DEFAULT);
		boolean rewrite = rewriteP.orElse(false);
		boolean breakOnExcept = breakOnExceptP.orElse(false);
		boolean genEmptyFile = genEmptyFileP.orElse(false);
		String enc = encP.orElse("UTF-8");

		resampleHisTickDataFileWind(windCodes, td, resFreq, cnFuNames, baseDir, rewrite, breakOnExcept, genEmptyFile, enc);
	}

	/**
	 * 仅支持 Tick 级别 resample 到日内级别: MIN_1, MIN_3, MIN_5, MIN_15, MIN_30, MIN_60:
	 * 
	 * @param windCodes
	 * @param td
	 * @param resFreq
	 * @param cnFuNames
	 * @param baseDir
	 * @param rewrite
	 * @param breakOnExcept
	 * @param genEmptyFile
	 * @param enc
	 * @throws Exception
	 */
	public static void resampleHisTickDataFileWind(List<String> windCodes, LocalDate td, Frequency resFreq,
	        List<String> cnFuNames, Path baseDir, boolean rewrite, boolean breakOnExcept, boolean genEmptyFile,
	        String enc) throws Exception {

		if (!Frequency.getIntradayFreqsCommon().contains(resFreq)) {
			throw new IllegalArgumentException(
			        String.format("Unsupported resample frequency (NOT intraday frequency): %s", resFreq));
		}

		Path tickCsvDir = baseDir.toAbsolutePath().resolve("tick").resolve(td.format(Constants.DATE_FORMAT_YYYY))
		        .resolve(td.format(Constants.DATE_FORMAT_YYYYMMDD));
		Path rsmpCsvDir = baseDir.toAbsolutePath().resolve(CommonDataUtils.barFreqToHisDataDir(resFreq))
		        .resolve(td.format(Constants.DATE_FORMAT_YYYY)).resolve(td.format(Constants.DATE_FORMAT_YYYYMMDD));
		if (!rsmpCsvDir.toFile().exists()) {
			rsmpCsvDir.toFile().mkdirs();
		}
		// windCodes 为 None 或 空集合, 则 resample 源 Tick 文件夹中所有品种:
		if (windCodes == null || windCodes.size() == 0) {
			windCodes = CommonDataUtils.getAllCodesOrWcsFromDir(tickCsvDir, Optional.of(cnFuNames), Optional.empty());
		}

		for (String e : windCodes) {
			try {
				Path aRsmpCsvFile = rsmpCsvDir.resolve(e + ".csv");
				if (!aRsmpCsvFile.toFile().exists() || rewrite) {
					// 读取 td 交易日所有的 tick:
					Path aTickCsvFile = tickCsvDir.resolve(e + ".csv");
					// @formatter:off
					Map<String, ColumnType> colsTypeMap = MapUtils.putAll(new HashMap<>(), new Object[][] {
						{ Constants.DATE_TIME, ColumnType.LOCAL_DATE_TIME },
						{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("pre_close", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("last", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("volume", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("amt", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_WST.get("oi", "resp"), ColumnType.DOUBLE }
					});
					ColumnType[] types = CommonUtils.genColumnsInfoFor(aTickCsvFile, colsTypeMap, enc);
					// @formatter:on
					Table df;
					try (InputStreamReader reader = new InputStreamReader(new FileInputStream(aTickCsvFile.toFile()),
					        Charset.forName(enc))) {
						CsvReadOptions options = CsvReadOptions.builder(reader).missingValueIndicator("")
						        .columnTypes(types).dateTimeFormat(Constants.DATE_TIME_FORMAT_MS).build();

						df = Table.read().usingOptions(options);
					}
					// 过滤掉 datetime 不在交易时间范围内的 tick:
					IntArrayList indxes = new IntArrayList();
					df.forEach(row -> {
						if (CommonDataUtils.isIllegalCnFuTickDatetime(e, row.getDateTime(Constants.DATE_TIME),
								OptionalInt.empty(), Optional.empty(), Optional.empty(), Optional.empty())) {
							indxes.add(row.getRowNumber());
						}
					});
					if (indxes.size() > 0) {
						df = df.dropRows(indxes.toIntArray());
					}
					// 过滤重复 Tick (仅盘口有变动无成交的 Tick, 郑商所收盘 Tick, ...):
					double lstVol = 0.0;
					IntArrayList indxes1 = new IntArrayList();
					for (Row row : df) {
						double v = row.getDouble(
						        (String) Constants.WIND_FIELD_NAME_MAPPING_WST.get("volume", "resp"));
						if (DoubleMath.fuzzyEquals(v, lstVol, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH)) {
							indxes1.add(row.getRowNumber());
						}
						lstVol = v;
					}
					if (indxes1.size() > 0) {
						df = df.dropRows(indxes1.toIntArray());
					}

					if (df.isEmpty()) {
						continue;
					}
					// 用来计算 bar 的 涨跌 及 涨跌幅, 每新产生一个 bar 后更新, 初始值为前收盘价 (从 tickDf 中获取):
					double priceRefVal = df.row(0).getDouble(
					        (String) Constants.WIND_FIELD_NAME_MAPPING_WST.get("pre_close", "resp"));
					// 交易时间段等信息:
					Map<String, Object> cnFuTdInfos = CommonDataUtils.getTradeInfos(e, td);
					LocalDateTime startDm;
					LocalDateTime endDm;
					if (CommonDataUtils.isCnFuTdNightMarketAvailable(e, td)) {
						// 中国期货: 当前交易日包含上一交易日夜盘:
						startDm = LocalDateTime.of(tradingDaysOffset(td, 1), (LocalTime) cnFuTdInfos.get("openTimeNt"));
					} else {
						startDm = LocalDateTime.of(td, (LocalTime) cnFuTdInfos.get("openTimeAm"));
					}
					endDm = LocalDateTime.of(td, (LocalTime) cnFuTdInfos.get("closeTimePm"));
					// 收盘 (交易暂停) last tick 最大延迟时间 (秒):
					int lstTkDelay = CommonDataUtils.getCnFuLastTickDelay(e);
					// 对 Ticks 进行 sample 得到 min_1 级别 Bars:
					Table rsmpDf = sampleDataFrameTicksToMinuteWind(e, td, df, startDm, endDm, cnFuTdInfos, lstTkDelay);

					if (resFreq != Frequency.MINUTE && resFreq != Frequency.MINUTE_1) {
						// 其他分钟级别 Bar df, 由 1 分钟 Bar 经过resample 得到:
						rsmpDf = resampleDataFrameMinuteToWind(rsmpDf, startDm, endDm, resFreq, priceRefVal);
					}

					// 写入 csv:
					if (!rsmpDf.isEmpty() || genEmptyFile) {
						/*rsmpDf.write().csv(CsvWriteOptions.builder(aRsmpCsvFile.toFile()).build());*/
						CommonUtils.toCsv(rsmpDf, aRsmpCsvFile.toFile(), CommonUtils.CSV_FORMAT_DEFAULT, rewrite, enc);
						logger.info("resample (freq: {}) {} Ticks to {} (total: {} bars).", resFreq, e, aRsmpCsvFile,
						        rsmpDf.rowCount());
					}
				}
			} catch (Exception ex) {
				if (breakOnExcept) {
					throw ex;
				} else {
					String stkInfo = ExceptionUtils.getStackTrace(ex);
					logger.error(
					        "exception occured in resampleHisTickDataFileWind({}, {}, {}, {}, {}, {}, {}, {}, {}): {ex: {}, stackTrace: {}}",
					        windCodes, td, resFreq, cnFuNames, baseDir, rewrite, breakOnExcept, genEmptyFile, enc, ex,
					        stkInfo);
				}
			}
		}
	}

	/**
	 * 仅支持 MIN_1 级别 resample 到日内级别: MIN_3, MIN_5, MIN_15, MIN_30, MIN_60:
	 * 
	 * @param windCodes
	 * @param td
	 * @param resFreq
	 * @param cnFuNamesP default: Collections.emptyList()
	 * @param baseDirP default: WIND_HIS_DATA_BASE_DIR_DEFAULT
	 * @param rewriteP default: false
	 * @param breakOnExceptP default: false
	 * @param genEmptyFileP default: false
	 * @param dropEmptyBarsP default: true
	 * @param encP default: "UTF-8"
	 * @throws Exception
	 */
	public static void resampleHisMin1DataFileWind(List<String> windCodes, LocalDate td, Frequency resFreq,
	        Optional<List<String>> cnFuNamesP, Optional<Path> baseDirP, Optional<Boolean> rewriteP,
	        Optional<Boolean> breakOnExceptP, Optional<Boolean> genEmptyFileP, Optional<Boolean> dropEmptyBarsP,
	        Optional<String> encP) throws Exception {

		List<String> cnFuNames = cnFuNamesP.orElse(Collections.emptyList());
		Path baseDir = baseDirP.orElse(WIND_HIS_DATA_BASE_DIR_DEFAULT);
		boolean rewrite = rewriteP.orElse(false);
		boolean breakOnExcept = breakOnExceptP.orElse(false);
		boolean genEmptyFile = genEmptyFileP.orElse(false);
		boolean dropEmptyBars = dropEmptyBarsP.orElse(true);
		String enc = encP.orElse("UTF-8");

		resampleHisMin1DataFileWind(windCodes, td, resFreq, cnFuNames, baseDir, rewrite, breakOnExcept,
				genEmptyFile, dropEmptyBars, enc);
	}

	/**
	 * 仅支持 MIN_1 级别 resample 到日内级别: MIN_3, MIN_5, MIN_15, MIN_30, MIN_60:
	 * 
	 * @param windCodes
	 * @param td
	 * @param resFreq
	 * @param cnFuNames
	 * @param baseDir
	 * @param rewrite
	 * @param breakOnExcept
	 * @param genEmptyFile
	 * @param dropEmptyBars
	 * @param enc
	 * @throws Exception
	 */
	public static void resampleHisMin1DataFileWind(List<String> windCodes, LocalDate td, Frequency resFreq,
	        List<String> cnFuNames, Path baseDir, boolean rewrite, boolean breakOnExcept, boolean genEmptyFile,
	        boolean dropEmptyBars, String enc) throws Exception {

		if (resFreq == Frequency.MINUTE || resFreq == Frequency.MINUTE_1) {
			return;
		}

		if (!Frequency.getIntradayFreqsCommon().contains(resFreq)) {
			throw new IllegalArgumentException(
			        String.format("Unsupported resample frequency (NOT intraday frequency): %s", resFreq));
		}

		Path min1CsvDir = baseDir.toAbsolutePath().resolve(CommonDataUtils.barFreqToHisDataDir(Frequency.MINUTE))
		        .resolve(td.format(Constants.DATE_FORMAT_YYYY)).resolve(td.format(Constants.DATE_FORMAT_YYYYMMDD));
		Path rsmpCsvDir = baseDir.toAbsolutePath().resolve(CommonDataUtils.barFreqToHisDataDir(resFreq))
		        .resolve(td.format(Constants.DATE_FORMAT_YYYY)).resolve(td.format(Constants.DATE_FORMAT_YYYYMMDD));
		if (!rsmpCsvDir.toFile().exists()) {
			rsmpCsvDir.toFile().mkdirs();
		}
		// windCodes 为 None 或 空集合, 则 resample 源 min1 文件夹中所有品种:
		if (windCodes == null || windCodes.size() == 0) {
			windCodes = CommonDataUtils.getAllCodesOrWcsFromDir(min1CsvDir, Optional.of(cnFuNames), Optional.empty());
		}

		for (String e : windCodes) {
			try {
				Path aRsmpCsvFile = rsmpCsvDir.resolve(e + ".csv");
				if (!aRsmpCsvFile.toFile().exists() || rewrite) {
					//
					Path aMin1CsvFile = min1CsvDir.resolve(e + ".csv");
					// @formatter:off
					Map<String, ColumnType> colsTypeMap = MapUtils.putAll(new HashMap<>(), new Object[][] {
						{ Constants.DATE_TIME, ColumnType.LOCAL_DATE_TIME },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("open", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("close", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("high", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("low", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("volume", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("amt", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("chg", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("pct_chg", "resp"), ColumnType.DOUBLE },
						{ Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("oi", "resp"), ColumnType.DOUBLE },
					});
					ColumnType[] types = CommonUtils.genColumnsInfoFor(aMin1CsvFile, colsTypeMap, enc);
					// @formatter:on
					Table df;
					try (InputStreamReader reader = new InputStreamReader(new FileInputStream(aMin1CsvFile.toFile()),
					        Charset.forName(enc))) {
						CsvReadOptions options = CsvReadOptions.builder(reader).missingValueIndicator("")
						        .columnTypes(types).dateTimeFormat(Constants.DATE_TIME_FORMAT_STD).build();

						df = Table.read().usingOptions(options);
					}
					if (df.rowCount() == 0) {
						continue;
					}

					/* 
					 * 用来计算 bar 的 涨跌 及 涨跌幅, 每新产生一个 bar 后更新, 初始值为前收盘价 (从 min1 bars 中获取):
					 * 前收盘 = 首个 min1Bar close - priceChange
					 */
					double priceRefVal = df.row(0)
					        .getDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("close", "resp"))
					        - df.row(0).getDouble((String) Constants.WIND_FIELD_NAME_MAPPING_MIN_1.get("chg", "resp"));

					String exchId = Constants.exchSffxToName(CommonDataUtils.windCodeTuple(e).getRight());
					LocalDateTime startDm;
					LocalDateTime endDm;
					if (!Arrays.asList("SSE", "SZSE").contains(exchId)) { // 中国期货:
						// 交易时间段等信息:
						Map<String, Object> cnFuTdInfos = CommonDataUtils.getTradeInfos(e, td);
						if (CommonDataUtils.isCnFuTdNightMarketAvailable(e, td)) {
							// 中国期货: 当前交易日包含上一交易日夜盘:
							startDm = LocalDateTime.of(tradingDaysOffset(td, 1),
							        (LocalTime) cnFuTdInfos.get("openTimeNt"));
						} else {
							startDm = LocalDateTime.of(td, (LocalTime) cnFuTdInfos.get("openTimeAm"));
						}
						endDm = LocalDateTime.of(td, (LocalTime) cnFuTdInfos.get("closeTimePm"));
					} else { // 沪深交易所品种:
						startDm = LocalDateTime.of(td, Constants.OPEN_TIME_AM);
						endDm = LocalDateTime.of(td, Constants.CLOSE_TIME_PM);
					}

					// 其他分钟级别 Bar df, 由 1 分钟 Bar 经过resample 得到:
					Table rsmpDf = resampleDataFrameMinuteToWind(df, startDm, endDm, resFreq, priceRefVal,
					        dropEmptyBars);

					// 写入 csv:
					if (!rsmpDf.isEmpty() || genEmptyFile) {
						CommonUtils.toCsv(rsmpDf, aRsmpCsvFile.toFile(), CommonUtils.CSV_FORMAT_DEFAULT, rewrite, enc);
						logger.info("resample (freq: {}) {} Ticks to {} (total: {} bars).", resFreq, e, aRsmpCsvFile,
						        rsmpDf.rowCount());
					}
				}
			} catch (Exception ex) {
				if (breakOnExcept) {
					throw ex;
				} else {
					String stkInfo = ExceptionUtils.getStackTrace(ex);
					logger.error(
					        "exception occured in resampleHisMin1DataFileWind({}, {}, {}, {}, {}, {}, {}, {}, {}): {e: {}, stackTrace: {}}",
					        windCodes, td, resFreq, cnFuNames, baseDir, rewrite, breakOnExcept, genEmptyFile, enc, ex,
					        stkInfo);
				}
			}
		}
	}

	/**
	 * 
	 */
	private WindDataUtils() {

	}
}
